package org.lib.kafkahello;

import java.util.Arrays;
import java.util.Date;
import java.util.function.Consumer;
import java.util.stream.IntStream;

import org.lib.kafkahello.try1.Try1Config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import lombok.extern.slf4j.Slf4j;

@SpringBootApplication
public class KafkaHelloApplication {

	private static Consumer<Object> cs = System.out::println;
	
	public static void main(String[] args) {
		ConfigurableApplicationContext ctx = SpringApplication.run(KafkaHelloApplication.class, args);
		
		// 调试用
//		studyContext(ctx);
	}
	
	public static void studyContext(ConfigurableApplicationContext ctx) {
		cs.accept("++++++studyContext++++++\nBeanDefinitionCount=" + ctx.getBeanDefinitionCount());
		String[] beanNames = ctx.getBeanDefinitionNames();
		cs.accept("beanNames.length=" + beanNames.length);
		cs.accept("输出Bean的名称：");
		Arrays.stream(beanNames).forEach((name)->{
			cs.accept("name=" + name);
		});
		
//		cs.accept("---");
//		Arrays.stream(beanNames).forEach((name)->{
//			cs.accept(String.format("name=%s, bean=%s", name, ctx.getBean(name).toString()));
//		});
		
		cs.accept("------studyContext------");
	}

}

//@RestController
@Slf4j
class HomeController {

	@Autowired
	private KafkaTemplate<String, Object> kafkaTemplate;
	
	/**
	 * 调用接口发送消息
	 * 迁移到了 Try1Controller
	 * @author ben
	 * @date 2021-08-24 16:31:33 CST
	 * @return
	 */
	@GetMapping(value="/send")
	public Boolean sendMsg() {
		IntStream.rangeClosed(0, 9).forEach(i->{
			log.info("sendMsg-{}", i);
			String msg = String.format("HomeController send msg-%d @%s", i, new Date());
			
			// 无key
			kafkaTemplate.send(Try1Config.TOPIC_01, msg);
			
			// 有key
//			kafkaTemplate.send(Try1Config.TOPIC_01, Try1Config.TOPIC_01_KEY, "key-" + msg);
			
			// partition=0
//			kafkaTemplate.send(Try1Config.TOPIC_01, 0, Try1Config.TOPIC_01_KEY, "key-partition-0-" + msg);
			
			// partition=1 超过主题的分区数量
			// Error: 程序卡住，最后超时终止
			// org.apache.kafka.common.errors.TimeoutException: Topic topic01 not present in metadata after 60000 ms.
//			kafkaTemplate.send(Try1Config.TOPIC_01, 1, Try1Config.TOPIC_01_KEY, "key-partition-1-" + msg);
		});
		
		return true;
	}
	
}
